'''
@Company: TWL
@Author: xue jian
@Email: xuejian@kanzhun.com
@Date: 2020-07-27 14:11:19
'''
from transfer import *
from kafka import KafkaProducer
from kafka import KafkaConsumer
from kafka import TopicPartition
import json
if __name__ == "__main__":
    transfer = Transfer()
    bootstrap_servers=['172.21.211.126:9092','172.21.211.82:9092','172.21.211.41:9092','172.21.211.39:9092','172.21.211.214:9092']
    consumer_bootstrap_servers=['172.21.32.178:9092', '172.21.32.154:9092', '172.21.32.41:9092', '172.21.32.50:9092', '172.21.32.98:9092']
    topic = 'boss.arc.recommender.bossrec_sample_flow'
    consumer = KafkaConsumer(topic, auto_offset_reset='latest', group_id='transfer_special', bootstrap_servers=consumer_bootstrap_servers)
    producer = KafkaProducer(bootstrap_servers=bootstrap_servers)
    produce_topic = 'boss.arc_six.rcd.boss_recall_special_job'
    i = 0
    for msg in consumer:
        msg = json.loads(msg.value.decode())
        if msg['userFeatures'].get('1054', '30') == '30' or msg['listFeatures'].get('1053', '30') == '30':
            # log(json.dumps(msg))
            continue
        # print(msg)
        p_msg = transfer.transfer_b_data(msg)
        # if p_msg == None:
        #     # print('None')
        #     continue
        # try:
        #     # print(p_msg)
        #     producer.send(produce_topic, p_msg.encode(encoding='utf-8'))
        # except:
        #     log('transfer error!!!')
        #     continue
        # i += 1
        # if i > 6000000:
        #     log('consumer 2 million data')
        #     i = 0
